package com.winit.cloudlink.mgmt.recovery;

import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageListener;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;

import java.io.File;
import java.io.IOException;
import java.util.Timer;
import java.util.TimerTask;

@Slf4j
public class MessageBackupTask {

    private ConnectionFactory connectionFactory;
    private MessageWriter messageWriter;
    private String queueName;
    private File backupFile;

    private SimpleMessageListenerContainer container;

    public MessageBackupTask(ConnectionFactory connectionFactory, String queueName, String backupFile) {
        this.connectionFactory = connectionFactory;
        this.queueName = queueName;
        this.backupFile = new File(backupFile);
    }


    public void execute() {
        try {
            log.info("开始备份队列'{}'里的消息到文件'{}'.", queueName, backupFile);
            messageWriter = new FileMessageWriter(backupFile);
            container = new SimpleMessageListenerContainer();
            container.setConnectionFactory(connectionFactory);
            container.setQueueNames(queueName);
            container.setPrefetchCount(10);
            container.setTxSize(10);
            container.setConcurrentConsumers(1);
            container.setMessageListener(new BackupMessageListener(queueName));
            container.start();
        } catch (IOException e) {
            log.error("备份队列'{}'里的消息出错.", queueName, e);
        }
    }

    public void stop() {
        if (container != null && container.isRunning()) {
            container.stop();
        }
        log.info("已停止备份队列'{}'里的消息.", queueName);
    }


    public File getBackupFile() {
        return backupFile;
    }

    public class BackupMessageListener implements MessageListener {
        private String queueName;
        private long count = 0;

        public BackupMessageListener(String queueName) {
            this.queueName = queueName;
            Timer timer = new Timer();
            timer.schedule(new LogTimerTask(), 1000, 1000);
        }

        @Override
        public void onMessage(Message message) {
            count++;
            messageWriter.write(message);
        }

        public class LogTimerTask extends TimerTask {
            private long previous = 0;

            public void run() {
                if (count > previous) {
                    log.info("正在备份队列'{}'里的消息，已备份{}条消息.", queueName, count);
                    previous = count;
                }
            }
        }
    }
}
